package cn.com.soole.demo.callback;

import cn.com.soole.demo.client.StuMqttClient;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.joda.time.DateTime;
import org.joda.time.Days;
import org.joda.time.Seconds;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.EnvironmentAware;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

import java.time.LocalDateTime;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * MQTT回调, 包括 客户端连接丢失回调, 订阅者消息到达回调, 发布者消息发布回调
 * @author create by jimmy 2116398125@qq.com
 * @package org.jimmy.mqtt.service
 * @date 2024/8/14 20:31
 */
@Slf4j
//@Configuration
public class MqttCallBackImpl implements MqttCallbackExtended {
    private final static AtomicInteger atom = new AtomicInteger(0);

    //@Value("${spring.mqtt.client_id}")
    //private String clientId;

    private final StuMqttClient calClient;

    public MqttCallBackImpl(StuMqttClient client) {
        calClient = client;
    }

    /**
     * 当客户端丢失了对服务端的连接后,触发的回调方法
     * 1. 可以做一些资源清理
     * 2. 可以做个重连
     *
     * @date 2024/8/15 22:07
     * @param throwable 异常
     * @author create by jimmy 2116398125@qq.com
     */
    @Override
    public void connectionLost(Throwable throwable) {
        //System.out.println(clientId + "与服务器断开连接");
        throwable.printStackTrace();
        System.out.println("客户端与服务器断开连接, 将进行再次连接");
        calClient.connect();
    }

    /**
     * 此回调针对的订阅者
     * 订阅者订阅的消息到达上层应用(消费者端)之后, 触发的消息回调
     * 注意:
     * 该方法由mqtt客户端同步调用, 在此方法未正确返回之前, 不会发送ACK确认消息到broker,
     * 一旦该方法抛出了异常, 客户端将异常关闭, 当再次连接时, 所有QoS1, QoS2且客户端未进行ACK确认的消息,
     * 都将由broker再次发送到客户端
     * @date 2024/8/15 22:05
     * @param topic, mqttMessage
     * @author create by jimmy 2116398125@qq.com
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        int num = atom.incrementAndGet();
        int i = num % 100;
        if (i == 0) {
            printMsg(message, num);
        }
//        if (atom.incrementAndGet() == 10) {
//            System.out.println("消息内容:" + new String(message.getPayload()));
//            System.out.println("当前已收消息数量: " + atom.incrementAndGet() + " 最后消息到达时间:" + LocalDateTime.now());
//        }
//
//        if (atom.incrementAndGet() == 100) {
//            System.out.println("消息内容:" + new String(message.getPayload()));
//            System.out.println("当前已收消息数量: " + atom.incrementAndGet() + " 最后消息到达时间:" + LocalDateTime.now());
//        }
//
//        if (atom.incrementAndGet() == 1000) {
//            System.out.println("消息内容:" + new String(message.getPayload()));
//            System.out.println("当前已收消息数量: " + atom.incrementAndGet() + " 最后消息到达时间:" + LocalDateTime.now());
//        }
//
//        if (atom.incrementAndGet() == 2000) {
//            System.out.println("消息内容:" + new String(message.getPayload()));
//            System.out.println("当前已收消息数量: " + atom.incrementAndGet() + " 最后消息到达时间:" + LocalDateTime.now());
//        }
//        if (atom.incrementAndGet() == 3000) {
//            System.out.println("消息内容:" + new String(message.getPayload()));
//            System.out.println("当前已收消息数量: " + atom.incrementAndGet() + " 最后消息到达时间:" + LocalDateTime.now());
//        }
//        if (atom.incrementAndGet() == 4000) {
//            System.out.println("消息内容:" + new String(message.getPayload()));
//            System.out.println("当前已收消息数量: " + atom.incrementAndGet() + " 最后消息到达时间:" + LocalDateTime.now());
//        }
//        if (atom.incrementAndGet() == 5000) {
//            System.out.println("消息内容:" + new String(message.getPayload()));
//            System.out.println("当前已收消息数量: " + atom.incrementAndGet() + " 最后消息到达时间:" + LocalDateTime.now());
//        }
    }

    private void printMsg(MqttMessage message, Integer num) {
        System.out.println("消息内容:" + new String(message.getPayload()));
        String l = new String(message.getPayload());

        long l1 = Long.parseLong(l) - System.currentTimeMillis();
        System.out.println("当前已收消息数量: " + num + " 最后消息到达时间:" + LocalDateTime.now() + " 相差的秒数:" + l1 + "毫秒" + "\n\n");
    }

    /**
     * 此方法针对的是发布者
     * 消息发布者, 发布完成且收到ACK确认后的回调
     * QoS0: 消息被网络发出后触发一次
     * QoS1: 当publish收到broker的PUBACK消息后触发
     * QoS2: 当publish收到broker的PUBCOMP消息后触发
     *
     * @date 2024/8/15 22:04
     * @param iMqttDeliveryToken token
     * @author create by jimmy 2116398125@qq.com
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        IMqttAsyncClient client = iMqttDeliveryToken.getClient();
        log.info(client.getClientId() + "发布消息成功！");
    }

    @Override
    public void connectComplete(boolean b, String s) {

    }
}
